package worker

import (
	"fmt"
	"mapreduce-go/call"
	"mapreduce-go/common"
	"net"
	"net/rpc"
	"os"
	"strconv"
	"sync"
)

// 工作者 等待doTask任务或shutdown
type Worker struct {
	sync.Mutex
	Addr       string // 名称
	Map        func(string) []common.KeyValue
	Reduce     func(string, []string) string
	NTask      int // 任务总量
	Concurrent int // 当前worker并发执行的任务量
	Listener   net.Listener
}

// 任务执行函数 在有新的任务被分配给该master时 master调用该函数执行任务
func (wk *Worker) DoTask(args *call.DoTaskArgs, _ *struct{}) error {

	fmt.Println("worker : do task")

	// 任务数量加1
	wk.Lock()
	wk.NTask++
	wk.Unlock()

	switch args.Phase {
	case common.MapPhase:
		common.DoMap(args.JobName,
			args.TaskId,
			args.File,
			args.NumberOtherPhase,
			wk.Map)

	case common.ReducePhase:
		common.DoReduce(args.JobName,
			args.TaskId,
			common.GetMergeName(args.JobName, args.TaskId),
			args.NumberOtherPhase,
			wk.Reduce)
	}

	fmt.Println("worker : task is done " + args.JobName + "-" + strconv.Itoa(args.TaskId))

	return nil
}

// 启动worker与master进行连接
func RunWorker(masterAddr, me string,
	mapFunc func(string) []common.KeyValue,
	reduceFunc func(string, []string) string) {

	fmt.Println("worker : run worker")

	wk := new(Worker)

	wk.Addr = me
	wk.Map = mapFunc
	wk.Reduce = reduceFunc

	server := rpc.NewServer()

	server.Register(wk)
	os.Remove(me)

	lis, err := net.Listen("tcp", me)
	if err != nil {
		fmt.Println("error : ", err.Error())
	}

	wk.Listener = lis

	// 注册到master中
	wk.register(masterAddr)

	for {
		coon, err := wk.Listener.Accept()

		if err != nil {
			fmt.Println("worker : error : ", err.Error())
			break
		} else {
			go server.ServeConn(coon)
		}

	}
	wk.Listener.Close()
}

// 注册 相当于告知worker的存在
func (wk *Worker) register(masterAddr string) {
	args := new(call.RegisterArgs)
	args.WorkerAddr = wk.Addr

	// 调用master的注册函数
	ok := call.Call(masterAddr, "Master.Register", args, nil)

	if !ok {
		fmt.Println("error can not connect master")
	} else {
		fmt.Println("worker : connected")
	}
}

func (wk *Worker) Shutdown(_ *struct{}, res *call.ShutdownReply) error {

	wk.Lock()
	defer wk.Unlock()

	res.Id = wk.NTask

	return nil
}
